package com.example.service;

import io.etcd.jetcd.*;
import io.etcd.jetcd.cluster.MemberListResponse;
import io.etcd.jetcd.kv.DeleteResponse;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.lease.LeaseRevokeResponse;
import io.etcd.jetcd.lease.LeaseTimeToLiveResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.LeaseOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.options.WatchOption;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;

@Service
public class EtcdServiceImpl implements EtcdService {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private Client client;

    @Override
    public Client getClient() {
        return this.client;
    }

    /***
     * 判断某个Key是否存在
     * @param key
     * @return
     * @throws Exception
     */
    @Override
    public boolean isExist(String key) throws Exception {
        return this.client.getKVClient().get(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8))).get().getKvs().contains(key);
    }

    /****
     * 异步设置key并返回结果
     * @param key
     * @param value
     * @return
     */
    @Override
    public CompletableFuture<PutResponse> putAsync(String key, byte[] value) {
        return this.client.getKVClient().put(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8)), ByteSequence.from(value));
    }

    /****
     * 同步设置key
     * @param key
     * @param value
     * @throws Exception
     */
    @Override
    public void putSync(String key, byte[] value) throws Exception {
        this.client.getKVClient().put(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8)), ByteSequence.from(value)).get();
    }


    /***
     * 异步删除key信息
     * @param key
     * @return
     */
    @Override
    public CompletableFuture<DeleteResponse> deleteAsync(String key) {
        return this.client.getKVClient().delete(ByteSequence.from(key, StandardCharsets.UTF_8));
    }


    /****
     * 同步删除key信息
     * @param key
     * @throws Exception
     */
    @Override
    public void deleteSync(String key) throws Exception {
        this.client.getKVClient().delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
    }

    /***
     * 设置key value和过期时间
     * @param key
     * @param value
     * @param ttlSeconds
     * @throws Exception
     */
    @Override
    public long putByTls(String key, byte[] value, long ttlSeconds) throws Exception {
        long grantId = this.client.getLeaseClient().grant(ttlSeconds).get().getID();
        PutOption putOption = PutOption.newBuilder().withLeaseId(grantId).build();
        this.client.getKVClient().put(ByteSequence.from(key, StandardCharsets.UTF_8), ByteSequence.from(value), putOption).get();
        return grantId;
    }

    /***
     * 设置kv的值
     * @param key
     * @param value
     * @throws Exception
     */
    @Override
    public void put(String key, byte[] value) throws Exception {
        this.client.getKVClient().put(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8)), ByteSequence.from(value)).get();
    }


    /***
     * 设置kv并自动续约
     * @param key
     * @param value
     * @throws Exception
     */
    @Override
    public void putAutoLease(String key, byte[] value, long ttlSeconds) throws Exception {
        long grantId = this.client.getLeaseClient().grant(ttlSeconds).get().getID();
        PutOption putOption = PutOption.newBuilder().withLeaseId(grantId).build();
        this.client.getKVClient().put(ByteSequence.from(key, StandardCharsets.UTF_8), ByteSequence.from(value), putOption).get();
        this.keepAlive(grantId);
    }

    /****
     * 获取某个key对应的value，如果key不存在就会报错的。
     * @param key
     * @return
     * @throws Exception
     */
    @Override
    public CompletableFuture<GetResponse> getAsync(String key) {
        return this.client.getKVClient().get(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8)));
    }

    /***
     * 同步获取消息
     * @param key
     * @return
     * @throws Exception
     */
    @Override
    public List<KeyValue> getSync(String key) throws Exception {
        return this.client.getKVClient().get(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8))).get().getKvs();
    }


    /****
     * 获取前缀对应的信息
     * @param key
     * @return
     * @throws Exception
     */
    @Override
    public List<KeyValue> getChildList(String key) throws Exception {
        ByteSequence byteSequence = ByteSequence.from(key.getBytes(StandardCharsets.UTF_8));
        GetOption getOption = GetOption.newBuilder().withPrefix(byteSequence).build();
        return this.client.getKVClient().get(byteSequence, getOption).get().getKvs();
    }


    /***
     * 获取续约信息
     * @param grantId
     * @return
     * @throws Exception
     */
    @Override
    public LeaseTimeToLiveResponse getKeyLease(long grantId) throws Exception {
        LeaseOption leaseOption = LeaseOption.newBuilder().withAttachedKeys().build();
        return this.client.getLeaseClient().timeToLive(grantId, leaseOption).get();
    }

    /***
     * 获取集群节点的列表
     * @return
     * @throws Exception
     */
    @Override
    public CompletableFuture<MemberListResponse> getMemberList() {
        return this.client.getClusterClient().listMember();
    }


    /***
     * 给key续约一次
     * @param grantId
     * @param grantId
     * @throws Exception
     */
    @Override
    public CompletableFuture<LeaseKeepAliveResponse> keepAliveOnce(long grantId) {
        return this.client.getLeaseClient().keepAliveOnce(grantId);
    }


    /***
     * 给key续约开启自动续约
     * @param grantId
     * @return
     * @throws Exception
     */
    @Override
    public void keepAlive(long grantId) throws Exception {
        this.client.getLeaseClient().keepAlive(grantId, new StreamObserver<LeaseKeepAliveResponse>() {
            @Override
            public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {

            }
        });
    }


    /***
     * 取消续约信息
     * @param grantId
     * @return
     * @throws Exception
     */
    @Override
    public CompletableFuture<LeaseRevokeResponse> revokeAsync(long grantId) throws Exception {
        return this.client.getLeaseClient().revoke(grantId);
    }


    /****
     * 取消续约信息
     * @param grantId
     * @return
     * @throws Exception
     */
    @Override
    public LeaseRevokeResponse revokeSync(long grantId) throws Exception {
        return this.client.getLeaseClient().revoke(grantId).get();
    }

    /****
     * 监听key信息的变化
     * @param key
     * @param listener
     * @return
     */
    @Override
    public Watch.Watcher watchKey(String key, Watch.Listener listener) {
        ByteSequence byteKey = ByteSequence.from(key, StandardCharsets.UTF_8);
        return this.client.getWatchClient().watch(byteKey, listener);
    }

    /****
     * 监听key目录信息的变化
     * @param key
     * @param listener
     * @return
     */
    @Override
    public Watch.Watcher watchPrefKey(String key, Watch.Listener listener) {
        ByteSequence byteKey = ByteSequence.from(key, StandardCharsets.UTF_8);
        return this.client.getWatchClient().watch(byteKey, WatchOption.newBuilder().withPrefix(byteKey).withPrevKV(true).build(), listener);
    }

    /***
     * 获取事务管理器
     * @return
     * @throws Exception
     */
    @Override
    public Txn getTxn() throws Exception {
        return this.client.getKVClient().txn();
    }
}
